package com.fzk.core.btree;

import com.fzk.log.Logger;
import com.fzk.utils.MyFileUtil;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 基于文件的B+树，实现了LRU缓存控制
 * TODO 改为维护最大key
 *
 * @author fzk
 * @datetime 2023-05-12 23:24:00
 */
public class BTreeFile implements AutoCloseable {
    private final Path path;
    private FileChannel fileChannel;
    private final FileHeader fileHeader;
    private final PageCache cache;// 1缓
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1);
    private final AtomicBoolean updateState = new AtomicBoolean(false);// 当前BTree是否处于更新状态
    // 常态: 0
    // 1 --> 提交落盘任务
    // 2 --> 落盘任务执行中
    private final AtomicInteger flushState = new AtomicInteger(0);
    private final ConcurrentHashMap<Integer, Page> updateMap = new ConcurrentHashMap<>();// 2缓, 存放更新操作过程中从lru缓存丢弃的页，必须暂存因为它们很可能还在被其它引用使用

    private class FileHeader {
        private static final int Page_Max_Num = 1 + Page.Key_Max_Num + Page.Key_Max_Num * Page.Key_Max_Num + Page.Key_Max_Num * Page.Key_Max_Num * Page.Key_Max_Num;// 限制4层高B+树
        public static final int BitSetLen = (Page_Max_Num + 1 + 7) / 8;// 位图的长度(字节数)，+1是因为第0页不使用非法
        public static final int File_Header_Size = BitSetLen + 4 + 4 + 4;// 文件头长度
        private BitSet bitSet;// 页面使用标记位图
        private int rootPageIdx;// 根结点页下标
        private int headPageIdx;// 叶子双向链表头页下标
        private int pageCount = 0;// 已使用页数量

        public ByteBuffer serialize() {
            ByteBuffer buffer = ByteBuffer.allocate(File_Header_Size);
            byte[] bitBytes = bitSet.toByteArray();
            buffer.put(bitBytes);
            if (bitBytes.length < BitSetLen) {
                byte[] padding = new byte[BitSetLen - bitBytes.length];
                buffer.put(padding);
            }
            buffer.putInt(rootPageIdx);
            buffer.putInt(headPageIdx);
            buffer.putInt(pageCount);
            buffer.flip();// 翻转待读
            return buffer;
        }

        public void clear() {
            this.bitSet.clear();
            this.rootPageIdx = Page.Invalid_PageIdx;
            this.headPageIdx = Page.Invalid_PageIdx;
            this.pageCount = 0;
        }

        public void flushHeader() {
            ByteBuffer headerBuf = serialize();
            long pos = 0L;
            while (headerBuf.hasRemaining()) {
                try {
                    pos += fileChannel.write(headerBuf, pos);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }

        @Override
        public String toString() {
            return "FileHeader{" +
                    "bitSet=" + bitSet +
                    ", rootPageIdx=" + rootPageIdx +
                    ", headPageIdx=" + headPageIdx +
                    ", pageCount=" + pageCount +
                    '}';
        }
    }

    public BTreeFile(Path path, int cachePageCount) throws IOException {
        this.path = path;
        MyFileUtil.ensureDirsAndFile(path);
        this.fileChannel = FileChannel.open(path, Set.of(StandardOpenOption.READ, StandardOpenOption.WRITE));
        this.cache = new PageCache(cachePageCount) {
            @Override
            protected void release(Integer key, Page page) {
                // 如果出现并发会非常的危险，比如正在序列化，其它线程修改了已经序列化的属性字段，那将忽略，会出现数据不一致
                if (updateState.get()) {
                    updateMap.putIfAbsent(key, page);
                } else {
                    if (page.modified) {
                        flushPage(page);
                    }
                }
            }
        };
        this.fileHeader = new FileHeader();
        if (fileChannel.size() == 0) {
            // 初始化文件头
            this.fileHeader.bitSet = new BitSet();
            this.fileHeader.rootPageIdx = Page.Invalid_PageIdx;
            this.fileHeader.headPageIdx = Page.Invalid_PageIdx;
            this.fileHeader.pageCount = 0;// 已使用页数量
        } else {
            // 从文件初始化文件头
            ByteBuffer buffer = ByteBuffer.allocate(FileHeader.File_Header_Size);
            do {
                int readLen = fileChannel.read(buffer, buffer.position());
                if (readLen == -1) {
                    throw new RuntimeException(String.format("超出文件范围了? fileSize: %d", fileChannel.size()));
                }
            } while (buffer.hasRemaining());

            buffer.flip();// 翻转待读
            // 读取位图
            byte[] bitBytes = new byte[FileHeader.BitSetLen];
            buffer.get(bitBytes);
            this.fileHeader.bitSet = BitSet.valueOf(bitBytes);
            this.fileHeader.rootPageIdx = buffer.getInt();
            this.fileHeader.headPageIdx = buffer.getInt();
            this.fileHeader.pageCount = buffer.getInt();
        }
    }

    /**
     * @param key 查询的key，不能为空
     * @return 有则返回，没有则为null
     */
    public Long get(String key) throws IOException {
        if (key == null || key.length() == 0) {
            throw new IllegalArgumentException("key不能为空");
        }
        rwLock.readLock().lock();
        try {
            // 如果还未初始化
            if (fileHeader.rootPageIdx <= Page.Invalid_PageIdx) {
                myAssert(fileHeader.headPageIdx <= Page.Invalid_PageIdx && fileHeader.pageCount == 0, String.format("rootPageIdx为0，但是headerPageIdx为: %s, pageCount: %d", fileHeader.headPageIdx, fileHeader.pageCount));
                return null;
            }
            Page root = getPage(fileHeader.rootPageIdx);
            Page leafPage = findLeafPageByKey(root, key);
            return leafPage.get(key);
        } finally {
            rwLock.readLock().unlock();
        }
    }

    /**
     * 获取此前缀匹配的总数量
     * TODO 优化: 从索引页开始匹配
     *
     * @param prefix 前缀, 这个不要在写成`abc*`的形式，要个`*`去掉
     * @return 匹配数量
     */
    public int prefixSearchCount(String prefix) {
        if (prefix == null || prefix.length() == 0) {
            throw new IllegalArgumentException("前缀模糊查询前缀不能为空");
        }
        rwLock.readLock().lock();
        try {
            int count = 0;
            // 1.找到开始搜索的叶子节点
            // 如果还未初始化
            if (fileHeader.rootPageIdx <= Page.Invalid_PageIdx) {
                myAssert(fileHeader.headPageIdx <= Page.Invalid_PageIdx && fileHeader.pageCount == 0, String.format("rootPageIdx为0，但是headerPageIdx为: %s, pageCount: %d", fileHeader.headPageIdx, fileHeader.pageCount));
                return count;
            }
            Page root = getPage(fileHeader.rootPageIdx);
            Page curPage = findLeafPageByKey(root, prefix);
            // 获取prefix在此叶子页存在时的槽索引或不存在时应该插入的槽索引, 总之从其开始都是满足的
            int curIndex = curPage.searchKeyIndex(prefix).index;
            // 先根据下一页的minKey来判断是否当前页都是匹配的
            while (curPage.hasNextPage()) {
                Page nextPage = getPage(curPage.nextPageIdx);
                if (nextPage.minKey().startsWith(prefix)) {
                    count += (curPage.keyCount - curIndex);// 这一页剩下的key都匹配
                    curIndex = 0;
                    curPage = nextPage;
                } else {
                    break;
                }
            }
            // 计算最后匹配的这页有多少个key匹配
            while (curIndex < curPage.keyCount) {
                if (curPage.entries[curIndex].key.startsWith(prefix)) {
                    count++;
                } else break;
                curIndex++;
            }
            return count;
        } catch (IOException e) {
            Logger.error(String.format("b+树出现了io出错，前缀搜索出错, prefix: %s, io err: %s", prefix, e));
            throw new RuntimeException(e);
        } finally {
            rwLock.readLock().unlock();
        }

    }

    /**
     * 前缀模糊查找
     *
     * @param prefix 前缀, 这个不要在写成`abc*`的形式，要个`*`去掉
     * @param offset 偏移量
     * @param limit  查询最多条数
     * @return 没有则为空集合, 不会返回null
     */
    public ArrayList<Page.BEntry> prefixSearch(String prefix, int offset, int limit) {
        if (prefix == null || prefix.length() == 0) {
            throw new IllegalArgumentException("前缀模糊查询前缀不能为空");
        }
        if (offset < 0 || limit <= 0) {
            throw new IllegalArgumentException(String.format("前缀prefix: %s 模糊查询参数错误, offset: %d, limit: %d", prefix, offset, limit));
        }
        rwLock.readLock().lock();
        try {
            ArrayList<Page.BEntry> result = new ArrayList<>();
            // TODO 计算total，提前比较
            // 1.找到开始搜索的叶子节点
            // 如果还未初始化
            if (fileHeader.rootPageIdx <= Page.Invalid_PageIdx) {
                myAssert(fileHeader.headPageIdx <= Page.Invalid_PageIdx && fileHeader.pageCount == 0, String.format("rootPageIdx为0，但是headerPageIdx为: %s, pageCount: %d", fileHeader.headPageIdx, fileHeader.pageCount));
                return result;
            }
            Page root = getPage(fileHeader.rootPageIdx);
            Page curPage = findLeafPageByKey(root, prefix);
            // 获取prefix在此叶子页存在时的槽索引或不存在时应该插入的槽索引, 总之从其开始都是满足的
            int curIndex = curPage.searchKeyIndex(prefix).index;
            // 2.先跳过offset个
            if (offset > 0) {
                int needSkipped = offset;
                while (needSkipped > 0) {
                    // 就在这一页
                    if (curIndex + needSkipped < curPage.keyCount) {
                        curIndex += needSkipped;
                        needSkipped = 0;
                    } else {
                        // 跳过这页剩下的key
                        needSkipped -= (curPage.keyCount - curIndex);
                        if (!curPage.hasNextPage()) {// 没有下一页则直接返回
                            return result;
                        }
                        curIndex = 0;// 下一页第一个key
                        curPage = getPage(curPage.nextPageIdx);// 下一页
                    }
                }
            }
            // 3.匹配key
            for (int i = 0; i < limit; i++) {
                if (curIndex >= curPage.keyCount) {
                    //  这页匹配完了，下一页
                    if (!curPage.hasNextPage()) {
                        return result;// 没有下一页直接返回
                    }
                    curIndex = 0;// 下一页第一个key
                    curPage = getPage(curPage.nextPageIdx);// 下一页
                }
                String key = curPage.entries[curIndex].key;
                if (!key.startsWith(prefix)) {
                    // key不匹配说明已经没有可以匹配的key了则直接返回
                    return result;
                }
                long val = curPage.entries[curIndex].val;
                result.add(new Page.BEntry(key, val));
                curIndex++;
            }
            return result;
        } catch (IOException e) {
            Logger.error(String.format("b+树出现了io出错，前缀搜索出错, prefix: %s, offset: %d, limit: %d, io err: %s", prefix, offset, limit, e));
            throw new RuntimeException(e);
        } finally {
            rwLock.readLock().unlock();
        }
    }

    /**
     * 添加key到文件B+树内, 存在相同key时覆盖
     *
     * @param key 插入的key，不能为空
     * @param val 子页面页下标或文件存储元信息offset
     * @return 返回可能存在的旧值或null
     */
    public Long put(String key, long val) throws IOException {
        if (key == null || key.length() == 0) {
            throw new IllegalArgumentException("key不能为空");
        }
        rwLock.writeLock().lock();
        updateState.set(true);// 进入更新状态
        try {
            Page root;
            // 需要初始化根节点页面
            if (fileHeader.rootPageIdx <= Page.Invalid_PageIdx) {
                myAssert(fileHeader.headPageIdx <= Page.Invalid_PageIdx && fileHeader.pageCount == 0, String.format("rootPageIdx为0，但是headerPageIdx为: %s, pageCount: %d", fileHeader.headPageIdx, fileHeader.pageCount));
                root = newEmptyPage(Page.Leaf_Page_Type, Page.Invalid_PageIdx);
            } else {
                root = getPage(fileHeader.rootPageIdx);
            }
            // 1.找到待插入的叶子页
            Page leafPage = findLeafPageByKey(root, key);
            // 2.插入
            Long old;
            if (leafPage.canInsert(key)) {// 直接插入
                old = leafPage.put(key, val, this);
            } else {
                // 页分裂
                Page rightPage = splitPage(leafPage);
                // 判断插入哪一页
                old = key.compareTo(rightPage.minKey()) >= 0 ?
                        rightPage.put(key, val, this) :
                        leafPage.put(key, val, this);
            }
            return old;
        } finally {
            flushUpdateMap();// 必须立刻落盘更新2缓的脏页，避免其它线程后续的读操作看不见更新的问题
            updateState.set(false);// 退出更新状态
            rwLock.writeLock().unlock();
            // 提交刷盘任务
            submitFlushTask();
        }
    }

    /**
     * 从树中删除key并返回其值
     *
     * @param key 删除key
     * @return 如果存在则返回值，否则null
     */
    public Long remove(String key) throws IOException {
        if (key == null || key.length() == 0) {
            throw new IllegalArgumentException("key不能为空");
        }
        rwLock.writeLock().lock();
        updateState.set(true);// 进入更新态
        try {
            Page root;
            // 需要初始化跟节点页面
            if (fileHeader.rootPageIdx <= Page.Invalid_PageIdx) {// 树为空直接返回
                myAssert(fileHeader.headPageIdx <= Page.Invalid_PageIdx && fileHeader.pageCount == 0, String.format("rootPageIdx为0，但是headerPageIdx为: %s, pageCount: %d", fileHeader.headPageIdx, fileHeader.pageCount));
                return null;
            } else {
                root = getPage(fileHeader.rootPageIdx);
                // 比最小key都小则直接返回
                if (key.compareTo(root.minKey()) < 0) {
                    return null;
                }
            }
            // 找到需要删除key所在的叶子页
            Page leafPage = findLeafPageByKey(root, key);
            return deleteKeyFromPage(leafPage, key);
        } finally {
            flushUpdateMap();// 必须立刻落盘更新2缓的脏页，避免其它线程后续的读操作看不见更新的问题
            updateState.set(false);// 退出更新态
            rwLock.writeLock().unlock();
            // 提交刷盘任务
            submitFlushTask();
        }
    }

    // 从页面中删除某个key
    private Long deleteKeyFromPage(Page page, String key) throws IOException {
        // 1.找到key在页面的下标索引
        Page.SearchResult result = page.searchKeyIndex(key);
        if (!result.exists) {// 树中不存在key
            return null;
        }
        long val = page.entries[result.index].val;
        // 2.删除后页面为空，把页面也删了
        if (page.keyCount == 1) {
            // 2.0 先删除页面自己并从缓存中移除页面
            deletePage(page);
            // 2.1 如果该节点是root，则直接清空b+树
            if (page.pageIdx == fileHeader.rootPageIdx) {
                this.clear();// 清空B+树
                return val;
            }
            // 2.2 维护双向链表
            if (page.prePageIdx > Page.Invalid_PageIdx) {
                setNextPageIdx(page.prePageIdx, page.nextPageIdx);
            }
            if (page.nextPageIdx > Page.Invalid_PageIdx) {
                setPrePageIdx(page.nextPageIdx, page.prePageIdx);
            }
            // 如果头节点页面被删除，则改为其后继节点页
            if (fileHeader.headPageIdx == page.pageIdx) {
                fileHeader.headPageIdx = page.nextPageIdx;
            }

            // 2.3 从父页面中删除指向此页面的key
            Page parentPage = getPage(page.parentPageIdx);
            deleteKeyFromPage(parentPage, key);
            return val;
        }
        // 3.从页面中删除key
        System.arraycopy(page.entries, result.index + 1, page.entries, result.index, page.keyCount - result.index - 1);
        page.entries[--page.keyCount] = null;// 置空避免内存泄漏
        page.decreasePageUsedSize(key);// 减少页面已使用空间
        page.modified = true;// 标记页面修改了

        // 4.如果删除的key为页面的minKey，需要循环修改祖先结点指向此节点的key
        if (result.index == 0) {
            String newMinKey = page.minKey();
            minKeyUpdated(page, key, newMinKey);
        }

        // 5.页面的key数量减少，尝试合并页面
        tryMergePage(page);
        return val;
    }

    /**
     * 叶子结点需要合并的情况：
     * |509 906|
     * |509|  |906|
     * |509|  |906|
     * |509|  |906|
     * 索引结点需要合并的情况：
     * |427 944|
     * |427|  |944|
     * |427|  |944|
     * |427 509 625 906|  |944 945 946|
     * <p>
     * 合并叶子节点页面：尽量都向左合并，这样可以不修改Head结点，且不修改minKey
     *
     * @param page 叶子结点和旁边的合并，索引结点怎么合并呢？
     */
    private void tryMergePage(Page page) throws IOException {
        if (page.keyCount >= Page.Key_Max_Num / 2) return;
        Page prePage = page.prePageIdx > Page.Invalid_PageIdx ? getPage(page.prePageIdx) : null;
        if (prePage != null && (prePage.keyCount + page.keyCount) <= Page.Key_Max_Num) {
            // 优先将page合入左页面
            doMergePage(prePage, page);
            return;
        }
        Page nextPage = page.nextPageIdx > Page.Invalid_PageIdx ? getPage(page.nextPageIdx) : null;
        if (nextPage != null && (nextPage.keyCount + page.keyCount) <= Page.Key_Max_Num) {
            // 再考虑将右结点合入leaf
            doMergePage(page, nextPage);
            return;
        }

        // 此节点页没有兄弟节点，只能是root节点页
        if (page.prePageIdx <= Page.Invalid_PageIdx && page.nextPageIdx <= Page.Invalid_PageIdx) {
            myAssert(page.pageIdx == fileHeader.rootPageIdx, String.format("pageIdx: %d没有兄弟节点居然不是root页: %d", page.pageIdx, fileHeader.rootPageIdx));
            // 如果root节点页key数量为1，是索引页时，则循环向下降低层高，即删除没必要的索引页
            Page root = page;
            while (root.keyCount == 1 && root.isIndexPage()) {
                Page childPage = getPage((int) root.entries[0].val);
                childPage.parentPageIdx = Page.Invalid_PageIdx;// root页没有父节点
                childPage.modified = true;// 标记页面发生了修改
                deletePage(root);
                root = childPage;
                fileHeader.rootPageIdx = root.pageIdx;
            }
        }
    }

    /**
     * 将right页面合入left页面，从右向左合并可以减少key的移动次数，且减少minKey的修改，且避免了头节点修改
     * 但是可能right页的key数量更多时，虽然减少了移动次数，但是维护父子关系的开销大的多
     */
    private void doMergePage(Page left, Page right) throws IOException {
        Logger.debug(String.format("即将发生页合并, 页:%s 将合并到页: %s", right, left));
        // 1.将右边的key拷贝到左边，如果是索引页将维护父子关系，父-->子
        System.arraycopy(right.entries, 0, left.entries, left.keyCount, right.keyCount);
        left.keyCount += right.keyCount;
        left.pageUsedSize += (right.pageUsedSize - Page.Page_Header_Meta_Size);
        left.modified = true;// 标记页面已经修改

        // 如果是索引页，需要维护父子关系，子-->父
        if (right.isIndexPage()) {
            for (int i = 0; i < right.keyCount; i++) {
                int childPageIdx = (int) right.entries[i].val;
                setParentPageIdx(childPageIdx, left.pageIdx);
            }
        }

        // 2.维护双向链表
        left.nextPageIdx = right.nextPageIdx;
        if (right.nextPageIdx > Page.Invalid_PageIdx) {
            setPrePageIdx(right.nextPageIdx, left.pageIdx);
        }

        // 3.删除right页面：此时必有父页面
        deletePage(right);
        Page parentPage = getPage(right.parentPageIdx);
        deleteKeyFromPage(parentPage, right.minKey());
    }

    private void setParentPageIdx(int pageIdx, int parentPageIdx) throws IOException {
        Page page = tryGetPage(pageIdx);
        if (page != null) {// 位于缓存中
            page.parentPageIdx = parentPageIdx;
            page.modified = true;// 标记修改了
        } else {// 不在缓冲中就直接修改文件
            Page.directWriteParentPageIdx(fileChannel, calculatePagePos(pageIdx), parentPageIdx);
        }
    }

    private void setPrePageIdx(int pageIdx, int prePageIdx) throws IOException {
        Page page = tryGetPage(pageIdx);
        if (page != null) {// 位于缓存中
            page.prePageIdx = prePageIdx;
            page.modified = true;// 标记修改了
        } else {// 不在缓冲中就直接修改文件
            Page.directWritePrePageIdx(fileChannel, calculatePagePos(pageIdx), prePageIdx);
        }
    }

    private void setNextPageIdx(int pageIdx, int nextPageIdx) throws IOException {
        Page page = tryGetPage(pageIdx);
        if (page != null) {// 位于缓存中
            page.nextPageIdx = nextPageIdx;
            page.modified = true;// 标记修改了
        } else {// 不在缓冲中就直接修改文件
            Page.directWriteNextPageIdx(fileChannel, calculatePagePos(pageIdx), nextPageIdx);
        }
    }

    private Page splitPage(Page leftPage) throws IOException {
        if (leftPage.keyCount <= 2) {
            throw new RuntimeException(String.format("页面pageIdx:%d的keyCount:%d不足以页分裂", leftPage.pageIdx, leftPage.keyCount));
        }
        Logger.debug(String.format("即将页分裂, pageIdx: %d, keyCount: %d", leftPage.pageIdx, leftPage.keyCount));
        // 1.创建一个相同页类型/父页的空白页
        Page rightPage = newEmptyPage(leftPage.pageType, leftPage.parentPageIdx);
        // 2.将页面的一半数据拷贝到新的空白页中
        int from = leftPage.keyCount >> 1;
        int copyLen = leftPage.keyCount - from;
        System.arraycopy(leftPage.entries, from, rightPage.entries, 0, copyLen);
        // 将旧页面的置null，避免内存泄露
        Arrays.fill(leftPage.entries, from, leftPage.keyCount, null);

        // 3.维护两个页面页头
        leftPage.keyCount = from;
        rightPage.keyCount = copyLen;
        // 维护页面使用空间
        for (int i = 0; i < copyLen; i++) {
            byte[] bytes = rightPage.entries[i].key.getBytes(StandardCharsets.UTF_8);
            leftPage.decreasePageUsedSize(bytes);
            rightPage.increasePageUsedSize(bytes);
        }
        // 标记两个页面都发生了修改
        leftPage.modified = true;
        rightPage.modified = true;

        // 4.如果是索引页, 需要维护其子页面的父页面
        if (leftPage.isIndexPage()) {
            for (int i = 0; i < rightPage.keyCount; i++) {
                int childPageIdx = (int) rightPage.entries[i].val;
                setParentPageIdx(childPageIdx, rightPage.pageIdx);// 维护 子 --> 父
            }
        }

        // 5.维护双向链表
        int nextPageIdx = leftPage.nextPageIdx;
        leftPage.nextPageIdx = rightPage.pageIdx;

        rightPage.prePageIdx = leftPage.pageIdx;
        rightPage.nextPageIdx = nextPageIdx;

        // 有效的下一页则必须维护
        if (nextPageIdx > Page.Invalid_PageIdx) {
            setPrePageIdx(nextPageIdx, rightPage.pageIdx);
        }

        // 6.将新页面插入到父页面中
        Page parentPage;
        if (leftPage.parentPageIdx <= Page.Invalid_PageIdx) {// 分裂页为root页, 其父页面必须新建
            myAssert(leftPage.pageIdx == fileHeader.rootPageIdx, String.format("页父页下标<=0，必须为root页, pageIdx: %d, rootPageIdx: %d", leftPage.pageIdx, fileHeader.rootPageIdx));
            // 创建新root页
            parentPage = newEmptyPage(Page.Index_Page_Type, Page.Invalid_PageIdx);
            fileHeader.rootPageIdx = parentPage.pageIdx;
            // 这里需将旧的root页插入新root页
            doPutPage(parentPage, leftPage);
        } else {
            parentPage = getPage(leftPage.parentPageIdx);
        }
        doPutPage(parentPage, rightPage);
        return rightPage;
    }

    private void doPutPage(Page parentPage, Page page) throws IOException {
        String minKey = page.minKey();
        Long old;
        if (parentPage.canInsert(minKey)) {
            old = parentPage.put(minKey, page.pageIdx, this);// 维护父-->子
            page.parentPageIdx = parentPage.pageIdx;// 维护子-->父
        } else {
            // 分裂父页面
            Page rightParentPage = splitPage(parentPage);
            // 判断插入哪一页
            if (minKey.compareTo(rightParentPage.minKey()) >= 0) {
                old = rightParentPage.put(minKey, page.pageIdx, this);// 维护父-->子
                page.parentPageIdx = rightParentPage.pageIdx;// 维护子-->父
            } else {
                old = parentPage.put(minKey, page.pageIdx, this);// 维护父-->子
                page.parentPageIdx = parentPage.pageIdx;// 维护子-->父
            }
        }
        page.modified = true;
        if (old != null) {
            Logger.error(String.format("将页面插入父页面时，怎么会出现key已经存在于父节点呢? parentPage: %s, page: %s", parentPage.toString(), page.toString()));
            throw new RuntimeException("将页面插入父页面时，怎么会出现key已经存在于父节点呢?");
        }
    }

    /**
     * 从给定的页面开始向下查询，直至查到key所在或应该插入的叶子节点页面
     *
     * @param from 开始查询的页面
     * @param key  查询的key
     * @return key所在的叶子页面、或应该插入的叶子页
     * @throws IOException I/O错误
     */
    private Page findLeafPageByKey(Page from, String key) throws IOException {
        if (from.isLeafPage()) {
            return from;
        }
        if (from.keyCount < 1) {
            throw new RuntimeException(String.format("pageIdx: %d的页面明明已经没有key了为何还存在?", from.pageIdx));
        }

        Page curPage = from;
        while (curPage.isIndexPage()) {
            // 查找下一页
            Page.SearchResult result = curPage.searchKeyIndex(key);
            int slotIdx;
            if (result.exists) {
                slotIdx = result.index;
            } else if (result.index == 0) {
                // 说明key比此索引页最小key都小，则走第一个子页
                slotIdx = 0;
            } else {// 默认是key待插入索引的前一页
                slotIdx = result.index - 1;
            }
            // 获取下一页
            int nextPageIdx = (int) curPage.entries[slotIdx].val;
            curPage = getPage(nextPageIdx);
        }
        return curPage;
    }

    // 当页面的minKey改变时必须级联更新其父节点中指向此节点的key
    void minKeyUpdated(Page page, String oldMinKey, String newMinKey) throws IOException {
        Logger.debug(String.format("第%d页更新minKey, oldMinKey: %s, newMinKey: %s", page.pageIdx, oldMinKey, newMinKey));
        // 没有父节点则不用更新
        if (page.parentPageIdx <= Page.Invalid_PageIdx) {
            return;
        }

        // 更新父节点指向此页面的key
        Page parentPage = getPage(page.parentPageIdx);
        if (parentPage.replaceKey(oldMinKey, newMinKey) == 0) {
            minKeyUpdated(parentPage, oldMinKey, newMinKey);
        }
    }

    private Page tryGetPage(int pageIdx) {
        // 先从1缓获取
        Page page = cache.get(pageIdx);
        // 更新操作可以从2缓获取
        if (page == null && updateState.get()) {
            page = updateMap.get(pageIdx);
        }
        return page;
    }

    // 获取页面，不在缓存时从文件读取到缓存
    private Page getPage(int pageIdx) throws IOException {
        if (pageIdx <= Page.Invalid_PageIdx) {
            throw new RuntimeException(String.format("pageIdx: %d非法", pageIdx));
        }
        if (!fileHeader.bitSet.get(pageIdx)) {
            throw new RuntimeException(String.format("位图中该页未使用, pageIdx: %d", pageIdx));
        }
        // 1.从缓存读取
        Page page = tryGetPage(pageIdx);
        if (page == null) {
            synchronized (this) { // double check
                page = tryGetPage(pageIdx);
                if (page == null) {
                    // 2.没有则从文件读取
                    Logger.debug(String.format("将从文件读取页面pageIdx: %d", pageIdx));
                    page = readPage(pageIdx);
                    // 3.再放入缓存
                    Page old = cache.put(pageIdx, page);
                    if (old != null) {
                        throw new RuntimeException(String.format("本来缓存中不存在pageIdx: %d, 但是为什么又存在了呢?", pageIdx));
                    }
                }
            }

        }
        return page;
    }

    private Page readPage(int pageIdx) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(Page.Page_Size);
        long pagePos = calculatePagePos(pageIdx);
        do {
            int readLen = fileChannel.read(buffer, pagePos + buffer.position());
            if (readLen == -1) {
                throw new RuntimeException(String.format("超出文件范围了? pos: %d, count: %d, readLen: %d, pageIdx: %d", pagePos, buffer.position(), readLen, pageIdx));
            }
        } while (buffer.hasRemaining());

        buffer.flip();// 翻转待使用
        Page page = Page.deserialize(buffer);
        myAssert(page.pageIdx == pageIdx, String.format("准备读取的pageIdx: %d, 但从文件中读到的pageIdx: %d，并发操作了?还是没落盘呢?", pageIdx, page.pageIdx));
        return page;
    }

    private Page newEmptyPage(int pageType, int parentPageIdx) {
        if (fileHeader.pageCount >= FileHeader.Page_Max_Num) {
            throw new RuntimeException(String.format("无法创建更多页面, pageCount: %d已经超过最大页面数: %d", fileHeader.pageCount, FileHeader.Page_Max_Num));
        }
        // 找到第一个为0的bit，必须从第1位开始找，因为第0页非法
        int pageIdx = fileHeader.bitSet.nextClearBit(1);
        Logger.debug(String.format("创建空白页, pageIdx: %d, pageType:%d, parentPageIdx: %d", pageIdx, pageType, parentPageIdx));
        if (fileHeader.pageCount == 0) {
            // 创建根节点页
            myAssert(pageType == Page.Leaf_Page_Type && parentPageIdx <= Page.Invalid_PageIdx, String.format("根节点页面创建参数非法, pageType: %d 必须是叶子页:%d, parentPageIdx:%d必须是非法页下标:%d", pageType, Page.Leaf_Page_Type, parentPageIdx, Page.Invalid_PageIdx));
            if (fileHeader.rootPageIdx > Page.Invalid_PageIdx || fileHeader.headPageIdx > Page.Invalid_PageIdx) {
                throw new RuntimeException(String.format("无页面时rootPageIdx:%d或headPageIdx:%d非法", fileHeader.rootPageIdx, fileHeader.headPageIdx));
            }
            fileHeader.rootPageIdx = pageIdx;
            fileHeader.headPageIdx = pageIdx;
        }
        Page page = new Page(pageType, pageIdx, parentPageIdx);
        fileHeader.pageCount++;
        fileHeader.bitSet.set(pageIdx, true);// 标记使用该页

        page.modified = true;// 标记页为脏，未落盘

        Page old = cache.put(pageIdx, page);// 放入缓存中
        if (old != null) {
            throw new RuntimeException(String.format("本来缓存中不存在pageIdx: %d, 但是为什么又存在了呢?", pageIdx));
        }
        return page;
    }

    private void deletePage(Page page) {
        page.deleted = true;
        fileHeader.bitSet.set(page.pageIdx, false);// 从位图标记删除
        fileHeader.pageCount--;// 页数-1
        cache.remove(page.pageIdx);
        updateMap.remove(page.pageIdx);
    }

    private long calculatePagePos(int pageIdx) {
        return FileHeader.File_Header_Size// 文件头
                + (long) pageIdx * Page.Page_Size;
    }

    private void flushPage(Page page) {
        if (page.deleted) return;// 页面已经删除了不用管
        Logger.debug(String.format("脏页落盘, pageIdx: %d", page.pageIdx));
        // 写入文件
        ByteBuffer buffer = page.serialize();
        long pos = calculatePagePos(page.pageIdx);
        try {
            while (buffer.hasRemaining()) {
                pos += fileChannel.write(buffer, pos);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // 写完文件后标记未发生修改，因为该页面很可能还在被其它对象引用并使用，标记其修改已经落盘
        page.modified = false;
    }

    private void flushUpdateMap() {
        updateMap.forEach((k, page) -> {
            if (page.modified) {
                flushPage(page);
            }
        });
        updateMap.clear();
    }

    private class FLushTask implements Runnable {
        @Override
        public void run() {
            try {
                // 获取读锁，避免其它线程获取写锁并修改数据
                // 这样整个落盘期间数据是不会更改的
                rwLock.readLock().lockInterruptibly();
                // 设置flushState为2表示正在刷盘
                int expect_1 = flushState.get();
                if (expect_1 != 1) {
                    throw new RuntimeException(String.format("什么情况? 当前的flushState:%d 不为1?", flushState.get()));
                }
                if (!flushState.compareAndSet(expect_1, 2)) {
                    throw new RuntimeException(String.format("什么情况? 其它线程修改了flushState:%d???", flushState.get()));
                }
                Logger.debug("purge线程落盘开始");
                try {
                    // 1.落盘缓存中的脏页
                    Map<Integer, Page> snapshot = cache.kvMap();
                    for (Map.Entry<Integer, Page> entry : snapshot.entrySet()) {
                        Page page = entry.getValue();
                        if (page.modified) {
                            flushPage(page);
                        }
                    }
                    // 2.落盘map的脏页
                    for (Map.Entry<Integer, Page> entry : updateMap.entrySet()) {
                        Page page = entry.getValue();
                        if (page.modified) {
                            flushPage(page);
                        }
                    }
                    updateMap.clear();
                    // 3.落盘文件头
                    fileHeader.flushHeader();
                } finally {
                    // 重置刷盘状态
                    if (!flushState.compareAndSet(2, 0)) {
                        Logger.error(String.format("重置刷盘状态时flushState:%d不为2?", flushState.get()));
                        flushState.set(0);// 强行重置
                    }
                    rwLock.readLock().unlock();
                    Logger.debug("purge线程落盘结束");
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    // 注意：落盘和更新操作是互斥的
    private void submitFlushTask() {// 目的在于组提交，合并多个刷盘请求
        if (flushState.compareAndSet(0, 1)) {
            // 等待1s再落盘
            scheduledExecutor.schedule(new FLushTask(), 1, TimeUnit.SECONDS);
        }
    }

    // 清空b+树，删除所有key和页面
    public void clear() {
        rwLock.writeLock().lock();// 写锁
        try {
            /*
             * 两种调用：
             * 1.外部直接调用
             * 2.删除key后整个BTree树为空
             * */
            // 保证处于更新态
            boolean allReadyUpdateState = !updateState.compareAndSet(false, true);
            try {
                fileHeader.clear();
                //fileHeader.flushHeader();
            /*
             这里必须清空缓存，且updateState必须为true，它会将缓存的页面放入updateMap中
             然后直接清空updateMap即可，这样可避免冗余落盘
            */
                cache.clear();
                updateMap.clear();
                // 重置存储文件，因为文件在使用过程将越来越大，将其删除再新建得到小文件
                resetFile();
            } finally {
                if (!allReadyUpdateState) {
                    // 如果调用此方法之前不处于更新态，则应该是外部直接调用了clear方法
                    // 将更新态重置
                    updateState.set(false);
                }
            }
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    /**
     * 重置存储文件，因为文件在使用过程将越来越大，将其删除再新建得到小文件
     */
    private void resetFile() {
        rwLock.writeLock().lock();
        try {
            if (fileChannel != null) {
                Logger.debug("重置B+树存储文件，当前文件大小: " + fileChannel.size());
                fileChannel.close();
                Files.deleteIfExists(path);
                MyFileUtil.ensureDirsAndFile(path);
                fileChannel = FileChannel.open(path, Set.of(StandardOpenOption.READ, StandardOpenOption.WRITE));
                Logger.debug(String.format("重置B+树文件完成，新文件大小: %d, path: %s", fileChannel.size(), path));
            }
        } catch (IOException e) {
            Logger.error(String.format("重置文件失败, path: %s, i/o err: %s", path, e));
            throw new RuntimeException(e);
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    @Override
    public void close() {
        scheduledExecutor.shutdownNow();// 立刻关闭落盘线程
        cache.clear();
        fileHeader.flushHeader();
        updateMap.forEach((key, page) -> {
            if (page.modified) {
                flushPage(page);
            }
        });
        updateMap.clear();
        try {
            fileChannel.force(true);
            fileChannel.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return "BTreeFile{" +
//                "fileChannel=" + fileChannel +
                "fileHeader=" + fileHeader +
//                ", cache=" + cache +
//                ", rwLock=" + rwLock +
//                ", scheduledExecutor=" + scheduledExecutor +
//                ", updateState=" + updateState +
//                ", flushState=" + flushState +
//                ", updateMap=" + updateMap +
                '}';
    }


    public void printSelf() throws IOException {
        if (fileHeader.pageCount <= 0) {
            Logger.println("空树!!!");
            return;
        }
        Logger.println("=============================================================================================");
        Logger.println(this.toString());
        Queue<Page> queue = new LinkedList<>();
        Page rootPage = getPage(fileHeader.rootPageIdx);
        Logger.println(rootPage.getKeysString());// 跟节点页
        queue.add(rootPage);
        outer:
        while (queue.size() > 0) {
            int len = queue.size();// 此层节点数
            StringBuilder sb = new StringBuilder();
            while (len-- > 0) {
                Page remove = queue.remove();
                if (remove.pageType != Page.Index_Page_Type) {
                    break outer;
                }
                for (int i = 0; i < remove.keyCount; i++) {
                    int childPageIdx = (int) remove.entries[i].val;
                    Page childPage = getPage(childPageIdx);
                    queue.add(childPage);
                    sb.append(childPage.getKeysString()).append(';').append(' ');
                }
            }
            Logger.println(sb.toString());
        }
        Logger.println("=============================================================================================");
    }

    public void checkTree() {
        rwLock.readLock().lock();
        Logger.println("开始检查");
        try {
            // 检查文件头
            myAssert(fileHeader.bitSet.cardinality() == fileHeader.pageCount, String.format("位图中使用页数为%d, 但是文件头的pageCount-1:%d", fileHeader.bitSet.cardinality(), fileHeader.pageCount - 1));
            if (fileHeader.pageCount == 0) {
                myAssert(fileHeader.rootPageIdx <= Page.Invalid_PageIdx && fileHeader.headPageIdx <= Page.Invalid_PageIdx, String.format(
                        "树为空, 文件头非法, rootPageIdx: %d, headPageIdx: %d", fileHeader.rootPageIdx, fileHeader.headPageIdx));
                return;
            }
            if (fileHeader.pageCount == 1) {
                myAssert(fileHeader.rootPageIdx == fileHeader.headPageIdx, String.format("只有1页则rootPageIdx: %d 和 headPageIdx: %d必须想等", fileHeader.rootPageIdx, fileHeader.headPageIdx));
            }

            // 检查页
            Page rootPage = getPage(fileHeader.rootPageIdx);
            if (fileHeader.pageCount > 1 && rootPage.isLeafPage()) {
                throw new RuntimeException(String.format("根节点页应该是索引页，但此时为叶子页, pageIdx: %d", rootPage.pageIdx));
            }
            Queue<Page> queue = new LinkedList<>();
            ArrayList<Page> pages = new ArrayList<>(fileHeader.pageCount);
            ArrayList<Page> leafPages = new ArrayList<>();
            queue.add(rootPage);
            pages.add(rootPage);
            if (rootPage.isLeafPage()) {
                leafPages.add(rootPage);
            }
            int curLeafIndex = 0;
            while (queue.size() > 0) {
                int len = queue.size();// 此层个数
                while (len-- > 0) {
                    Page page = queue.remove();
                    // 页内自检
                    page.checkPage();
                    // 索引页
                    // 检查此节点的key指向的是子结点min key
                    // 检查此节点和子结点的父子关系是否正常
                    if (page.isIndexPage()) {
                        for (int i = 0; i < page.keyCount; i++) {
                            int childPageIdx = (int) page.entries[i].val;
                            Page childPage = getPage(childPageIdx);
                            myAssert(page.entries[i].key.equals(childPage.minKey()), String.format("pageIdx: %d的key: %s不是指向子节点的minKey: %s", page.pageIdx, page.entries[i].key, childPage.minKey()));
                            myAssert(childPage.parentPageIdx == page.pageIdx, String.format("父节点指向错误, pageIdx: %d", childPageIdx));
                            pages.add(childPage);
                            queue.add(childPage);
                            if (childPage.isLeafPage()) {
                                leafPages.add(childPage);
                            }
                        }
                    }

                    // 叶子页: 检查双向链表是否正常
                    if (page.isLeafPage()) {
                        Page prePage = null, nextPage = null;
                        if (curLeafIndex > 0) {
                            prePage = getPage(page.prePageIdx);
                        }
                        if (curLeafIndex < leafPages.size() - 1) {
                            nextPage = getPage(page.nextPageIdx);
                        }
                        // 检验头节点页
                        if (prePage == null) {
                            myAssert(page.pageIdx == fileHeader.headPageIdx && curLeafIndex == 0, String.format("前继节点为空的页子页必不是头节点页, pageIdx: %d, curLeafIndex: %d", page.pageIdx, curLeafIndex));
                        }
                        // 此页的最大key必须比下一页的minKey小
                        if (nextPage != null) {
                            myAssert(page.maxKey().compareTo(nextPage.minKey()) < 0, String.format("此页maxKey: %s大于下一页minKey: %s", page.maxKey(), nextPage.minKey()));
                        }

                        // 检查索引树指向的叶子结点和叶子结点链表顺序一致
                        if (prePage != null) {
                            myAssert(leafPages.get(curLeafIndex - 1).pageIdx == prePage.pageIdx, String.format("索引树指向的叶子结点和叶子结点链表顺序不一致,pageIdx: %d", page.pageIdx));
                            myAssert(prePage.nextPageIdx == page.pageIdx, String.format("前一页指向的下一页是%d而不是%d", prePage.nextPageIdx, page.pageIdx));
                        }
                        if (nextPage != null) {
                            myAssert(nextPage.pageIdx == leafPages.get(curLeafIndex + 1).pageIdx, "非法");
                            myAssert(nextPage.prePageIdx == page.pageIdx, String.format("下一页指向的前一页非法, page: %s, nextPage: %s", page, nextPage));
                        }
                        curLeafIndex++;
                    }
                }
            }
            // 检查文件头
            myAssert(fileHeader.pageCount == pages.size(), String.format("页面数量不符合, pageCount: %d, 读取的页数量: %d", fileHeader.pageCount, pages.size()));
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            Logger.println("结束检查");
            rwLock.readLock().unlock();
        }
    }

    public static void myAssert(boolean flag, String msg) {
        if (!flag) throw new RuntimeException(msg);
    }
}
